package edu.csl.study.spark.basic

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.parallel.immutable

object Core_accumulator_Scala_local {

  val rootDir = System.getProperty("user.dir")+ "\\testFile\\"

  def main(args: Array[String]): Unit = {

    //1、构建sparkConf对象 设置application名称和master地址  spark://centos20:7077
    val sparkConf: SparkConf = new SparkConf().setAppName("Mysql").setMaster("local[2]")

    //2、构建sparkContext对象,该对象非常重要，它是所有spark程序的执行入口
    // 它内部会构建  DAGScheduler和 TaskScheduler 对象
    val sc = new SparkContext(sparkConf)

    //设置日志输出级别
    sc.setLogLevel("warn")

    //3、读取数据文件
    val data: RDD[String] = sc.textFile(rootDir+"persons.txt")
    //累加器：创建accumulator并初始化为0
    val accumulator = sc.longAccumulator("count")
    var count = 1;

    var list =  scala.collection.mutable.ListBuffer(1,2,3)
    //变长
    val l2 = scala.collection.mutable.ListBuffer(1,2,3,4)
    //添加单个元素
    println("partitions = "+data.partitions.length)

    /***
     * 1.driver中定义的变量传输到work节点，work节点的数据会累加，但是不会返回到Driver，最后打印的count还是原数据。
     *   这里并不是传值和传引用的问题，送个Map类型数据也是一样的
     * 2.注意：driver中定义的变量传输到work节点，其实是需要实现序列号的， 数字、String等基本类型默认实现了序列化。
     *
     * 3.driver中定义的变量传输到work节点，和广播变量的区别：
     *   广播变量是每个节点一份，而driver中的变量
     *
     * 4.广播变量也需要实现序列化
     */
    data.foreach(line =>{
      accumulator.add(1)
       println(line)
      count = count+1;
      println(count)
    })

    println("-------------------accumulator.value----------------"+accumulator.value)
    println("--------------------count---------------"+count)
    println("----------------------l2-------------"+l2)


    //8、关闭sc
    sc.stop()



  }
}
